package server

import (
	"../../deps/lessgo/utils"
	"../base"
	"../net"
	"../state"
	"../store"
	"strings"
	"sync"
	"time"
)

var (
	proposals         = map[uint64]*store.NodeProposal{}
	proposal_servlock sync.Mutex
	watcherlock       sync.Mutex
	watches           = map[string]ProposalWatcher{} // map[path]*
	watchmq           = make(chan *WatcherQueue, 100000)
)

type Proposer int

type ProposalPromise struct {
	VerNow uint64
	VerSet uint64
}

type ProposalWatcher map[string]int64 // map[host]ttl

type WatcherQueue struct {
	Path  string
	Event string
	Rev   uint64
}

/** TODO func WatcherInitialize() {

    go func() {

        for q := range watchmq {

            if w, ok := watches[q.Key]; ok {
                for hostid, ttl := range w {

                    // delay clean watcher queue
                    if ttl < time.Now().Unix() {
                        watcherlock.Lock()
                        delete(w, hostid)
                        watcherlock.Unlock()
                        continue
                    }

                    if ip, ok := kp[hostid]; ok {
                        // Println("Send to", ip +":"+ port)
                        msg := map[string]string{
                            "action": "WatchEvent",
                            "path":   q.Key,
                            "event":  q.Event,
                        }
                        prbc.Send(msg, ip+":"+conf.Config.KeeperPort)
                    }
                }
            }
        }
    }()
}
*/

func (p *Proposer) Cmd(rq *base.PeerRequest, rp *base.Reply) error {

	switch string(rq.Method) {
	case "GET":
		ProposerNodeGet(rq, rp)
	// case "GETS":
	// 	ProposerNodeGets(rq, rp)
	case "SET":
		ProposerNodeSet(rq, rp, false)
	case "DEL":
		ProposerNodeSet(rq, rp, true)
		// case "KPRMEMSET":
		// 	CmdKprMemSet(rq, rp)
	}
	return nil
}

func ProposerNodeGet(rq *base.PeerRequest, rp *base.Reply) {

	var nodeGet base.DataNode
	if err := utils.JsonDecode(rq.Body, &nodeGet); err != nil {
		return
	}

	if node, e := store.NodeGet(nodeGet.Key); e == nil {

		rp.Type = base.ReplyString
		// if node.C == store.NodeDirMagic || node.P == "/" {
		// 	node.T = store.NodeTypeDir
		// 	node.C = ""
		// 	ns, _ := store.NodeList(nodeGet.Key)
		// 	node.N = ns
		// } else {
		// 	node.T = store.NodeTypeKey
		// }

		rp.Body, _ = utils.JsonEncode(node)
	}
}

func ProposerNodeGets(rq *base.PeerRequest, rp *base.Reply) {

	var nodeSet base.DataNode

	e := utils.JsonDecode(rq.Body, &nodeSet)
	if e != nil {
		return
	}

	keys := strings.Split(nodeSet.Key, ",")
	if rs, e := store.NodeGets(keys); e == nil {
		rp.Type = base.ReplyString
		rp.Body, _ = utils.JsonEncode(rs)
	}

	return
}

func ProposerNodeSet(rq *base.PeerRequest, rp *base.Reply, del bool) {

	var nodeSet base.DataNode
	var verset uint64
	promised := make(chan uint8, len(state.PeerIds)*2+2)
	accepted := make(chan uint8, len(state.PeerIds)*2+2)
	valued, unvalued := 0, 0

	if err := utils.JsonDecode(rq.Body, &nodeSet); err != nil {
		return
	}

	pl := &store.NodeProposal{
		Key: nodeSet.Key,
		Val: nodeSet.Value,
		Ttl: nodeSet.Ttl,
		//VerNow:   node.R,
		//VerSet:   verset,
		Valued:   0,
		Unvalued: 0,
	}

	if del {
		pl.Val, nodeSet.Value = store.NodeDelMagic, store.NodeDelMagic
	}

	//nodeEvent := store.EventNone

	node, _ := store.NodeGet(nodeSet.Key)
	if node.R > 0 {

		if node.C == nodeSet.Value && nodeSet.Ttl > 0 && nodeSet.Ttl != node.L {

			pl.Force = true
			pl.VerNow, pl.VerSet, verset = node.R, node.R, node.R

			goto AcceptSend
		}

		if (node.C == nodeSet.Value && node.L == nodeSet.Ttl) ||
			(len(nodeSet.PrevValue) > 0 && nodeSet.PrevValue != node.C) {
			//Println("same node", nodeSet.Key)
			rp.Type = base.ReplyString
			nodeReply := base.DataNode{
				Key:     node.P,
				Value:   node.C,
				Version: node.R,
			}
			if node.L > 0 {
				nodeReply.Ttl = node.L
			}
			rp.Body, _ = utils.JsonEncode(nodeReply)

			return
		}

		//nodeEvent = store.EventNodeDataChanged
	} else {

		//nodeEvent = store.EventNodeCreated
	}

	verset, _ = store.Cmd("INCRBY", "sys:keeper:ver", 1).Uint64()
	pl.VerNow = node.R
	pl.VerSet = verset
	// pl := &store.NodeProposal{
	// 	Key:      nodeSet.Key,
	// 	Val:      nodeSet.Value,
	// 	Ttl:      nodeSet.Ttl,
	// 	VerNow:   node.R,
	// 	VerSet:   verset,
	// 	Valued:   0,
	// 	Unvalued: 0,
	// }
	proposals[verset] = pl

	// promised := make(chan uint8, len(state.PeerIds)*2+2)

	// Acceptor.Prepare
	for _, v := range state.Peers {

		go func(promised chan uint8, pl *store.NodeProposal, ip, port string) {

			call := net.NewNetCall()

			call.Method = "Acceptor.Prepare"
			call.Args = pl
			call.Reply = new(ProposalPromise)
			call.Addr = ip + ":" + port

			net.RpcInstance.Call(call)

			_ = <-call.Status

			rs := call.Reply.(*ProposalPromise)
			if rs.VerNow == pl.VerNow {
				promised <- 1
			} else {
				promised <- 0
			}
			//fmt.Println("call     3")
		}(promised, pl, v.Addr, v.Port)
	}

	// valued := 0
	// unvalued := 0

WaitPromise:
	for {
		select {
		case s := <-promised:
			//fmt.Println("get          ", s)
			if s == 1 {
				valued++
				if 2*valued > len(state.PeerIds) {
					//fmt.Println("Valued")
					break WaitPromise
				}
			} else if s == 0 {
				unvalued++
				if 2*unvalued > len(state.PeerIds) {
					rp.Type = base.ReplyError
					rp.Body = "UnValued"
					return
				}
			} else {
				rp.Type = base.ReplyError
				return
			}

		case <-time.After(time.Second * 15):
			rp.Type = base.ReplyError
			return
		}
	}

AcceptSend:
	// Acceptor.Accept
	// accepted := make(chan uint8, len(state.PeerIds)*2+2)

	for _, v := range state.Peers {

		go func(accepted chan uint8, pl *store.NodeProposal, ip, port string) {

			call := net.NewNetCall()

			call.Method = "Acceptor.Accept"
			call.Args = pl
			call.Reply = new(base.Reply)
			call.Addr = ip + ":" + port

			net.RpcInstance.Call(call)

			_ = <-call.Status

			rs := call.Reply.(*base.Reply)

			//fmt.Println("Acceptor.Accept", rs)
			if rs.Type == base.ReplyOK {
				accepted <- 1
			} else {
				accepted <- 0
			}
		}(accepted, pl, v.Addr, v.Port)
	}

	valued = 0
	unvalued = 0

WaitAccept:
	for {
		select {
		case s := <-accepted:
			if s == 1 {
				valued++
				if 2*valued > len(state.PeerIds) {

					//watchmq <- &WatcherQueue{strings.Trim(nodeSet.Key, "/"), nodeEvent, 0}

					rp.Type = base.ReplyString
					nodeReply := base.DataNode{
						Key:     nodeSet.Key,
						Value:   nodeSet.Value,
						Version: verset,
					}
					if nodeSet.Ttl > 0 {
						nodeReply.Ttl = nodeSet.Ttl
					}
					rp.Body, _ = utils.JsonEncode(nodeReply)

					break WaitAccept
				}
			} else if s == 0 {
				unvalued++
				if 2*unvalued > len(state.PeerIds) {
					rp.Type = base.ReplyError
					rp.Body = "UnValued"
					return
				}
			} else {
				rp.Type = base.ReplyError
				rp.Body = "UnValued"
				return
			}
		case <-time.After(time.Second * 15):
			rp.Type = base.ReplyError
			rp.Body = "UnValued"
			return
		}
	}
}
